<?php
declare (strict_types = 1);

namespace app\logic;
use Exception;

class RedisLogic
{
    public $redis;

    protected $streamName = 'test_task_stream';
    protected $callbackQueueName = 'test_callback_queue';

    public static function ins() 
    {
        static $cls;
        if (!$cls) {
            $cls = new static();
        }
        return $cls;
    }

    private function __construct() 
    {
        $this->redis = new \Redis();
        $this->redis->connect(env('REDIS.HOST'), intval(env('REDIS.PORT')) );
        $this->redis->auth(env('REDIS.AUTH'));
        $this->redis->streamName = env('STREAMKEY.STREAM_TASK_KEY','test_task_stream'); 
        $this->redis->callbackQueueName = env('STREAMKEY.STREAM_CALLBACK_KEY','test_callback_queue'); 
    }

    /**
     * 添加 stream 任务
     * @param string $type
     * @param array $data
     * @param bool | string $callback 是否需要回调，如果是字符串，则指定回调队列名称
     * @return array
     */
    public function addTaskStream(string $type, array $data, $callback = true)
    {
        $data = [
            'id' => getGuid(), // 任务ID，可以根据需要生成唯一ID
            'type' => $type, // 任务类型
            'create_time' => time(), // 创建时间
            'data' => $data
        ];
        if($callback === true){
            $data['callback_queue'] = $this->callbackQueueName; // 如果需要回调队列，指定队列名称
        }elseif($callback!== false){
            $data['callback_queue'] = $callback; // 如果需要回调队列，指定队列名称
        }
        $message = ['data'=> json_encode($data,JSON_UNESCAPED_UNICODE)];
        // 将消息发送到Redis Stream
        try {
            $result = $this->redis->xadd($this->streamName, '*', $message);
            return ["code"=>1,"msg"=>"消息已发送到Stream，消息ID: " . $result];
        } catch (Exception $e) {
            return ["code"=>0,"msg"=>"发送消息失败: " . $e->getMessage()];
        }
    }

    /**
     * 读取callback stream任务
     * @return void
     */
    public function callbackStream()
    {
        
        $lastId = '0-0'; // 初始读取位置用 0-0 表示从第一条开始读取

        while (true) {
            try {
                // 使用 XREAD 读取最新消息（阻塞式，超时 0 表示无限等待）
                $callbackData = $this->redis->xread(
                    [$this->callbackQueueName => $lastId], // Stream 名称 + 起始 ID
                    1,                       // COUNT 1（每次读取1条）
                    2                        // BLOCK 0（无限阻塞等待）
                );
                
                if (isset($callbackData[$this->callbackQueueName])) {
                    foreach ($callbackData[$this->callbackQueueName] as $key => $value) {
                        $messageId = $key;
                        $messageData = $value['message']; // 取出消息内容 是json数据
                        // 处理消息...
                        echo "收到新消息 ID: $messageId, 内容: " . $messageData. "\n";

                        // 更新 lastId，避免重复读取
                        $lastId = $messageId;
                        $this->redis->xdel("callback_queue", [$messageId]); // 删除已处理的消息
                    }
                } else {
                    // echo "没有新消息\n";
                    break;
                }
            } catch (Exception $e) {
                // echo "读取消息失败: " . $e->getMessage() . "\n";
                sleep(1); // 避免频繁重试
            }
        }
    }

}